/* tqueue.c
 * This file is part of the Tisa Project.
 * 
 * Authors: Cleiber Marques da Silva <cleiber.dev@gmail.com>
 *          Odilson Tadeu Valle <odilson@sj.cefetsc.edu.br>
 *
 * Tisa Project is free software; you can redistribute
 * it and/or modify it under the terms of the GNU General Public
 * License as published by the Free Software Foundation; version 2 of
 * the License.
 */

#ifdef HAVE_CONFIG_H
#include <config.h>
#endif
#include <time.h>
#include <string.h>
#include <unistd.h>
#include <errno.h>
#include <pthread.h>

#include "list.h"
#include "log.h"
#include "tqueue.h"

LIST_HEAD(tq_list);

static volatile int killed = 0;
static pthread_mutex_t mutex;
static pthread_cond_t cond;
static pthread_t tq_runner;

static volatile struct tq_elem *curr = NULL;

static inline int is_first_task(struct tq_elem *tqe)
{
    return (tq_list.next == &tqe->list);
}

static void *runner(void *arg);

/* taskqueue_init - Initializes task queue and creates a task runner thread */
int taskqueue_init(void)
{
	pthread_mutexattr_t mattrs;
	pthread_mutexattr_init(&mattrs);
	pthread_mutexattr_settype(&mattrs, PTHREAD_MUTEX_FAST_NP);
	if (pthread_mutex_init(&mutex, &mattrs) ||
	    pthread_cond_init(&cond, NULL) ||
	    pthread_create(&tq_runner, NULL, runner, NULL))
		return -1;
	return 0;
}

/* taskqueue_destroy - destroy task queue */
void taskqueue_destroy(void)
{
	struct list_head *l, *n;
	pthread_mutex_lock(&mutex);
	list_for_each_safe(l, n, &tq_list) {
		struct tq_elem *tqe;
		list_del(l);
		tqe = list_entry(l, struct tq_elem, list);
		tsclear(tqe->expires);
	}
	killed = 1;
	pthread_cond_signal(&cond);
	pthread_mutex_unlock(&mutex);
	pthread_join(tq_runner, NULL);
}

/* sorted_insert - insert queue item sorted by expiration time */
static inline void sorted_insert(struct tq_elem *tqi)
{
	struct list_head *l;

	list_for_each(l, &tq_list) {
		struct tq_elem *cur = list_entry(l, struct tq_elem, list);
		if (tscmp(cur->expires, tqi->expires, >)) {
			list_add_tail(&tqi->list, &cur->list);
			return;
		}
	}
	list_add_tail(&tqi->list, &tq_list);
}

/* add_task_abs - add new task */
int add_task_abs(const struct timespec *expires,
		 struct tq_elem *tqi, void (*task)(struct tq_elem *))
{
	pthread_mutex_lock(&mutex);
	if (tsisset(tqi->expires)) {
		list_del(&tqi->list);
	}
	if (tqi == curr)
		curr = NULL;
	tqi->expires = *expires;
	tqi->task = task;
	tqi->thread = pthread_self();
	sorted_insert(tqi);
	if (is_first_task(tqi))
		pthread_cond_signal(&cond);

	pthread_mutex_unlock(&mutex);

	return 0;
}

/* runner - Run expiring tasks */
static void *runner(void *arg)
{
	log_print(LOG_NOTICE, "thread runner started\n");

	pthread_mutex_lock(&mutex);
	for (;;) {
		struct timespec now;

		if (killed)
			break;

		if (list_empty(&tq_list))
			pthread_cond_wait(&cond, &mutex);

		clock_gettime(CLOCK_REALTIME, &now);

		while (!list_empty(&tq_list)) {
			struct tq_elem *first;
			first = list_entry(tq_list.next, struct tq_elem, list);
			if (tscmp(first->expires, now, >)) {
				pthread_cond_timedwait(&cond, &mutex,
						       &first->expires);
				break;
			}
			list_del(&first->list);
			tsclear(first->expires);
			curr = first;
			pthread_mutex_unlock(&mutex);
			first->task(first);
			pthread_mutex_lock(&mutex);
			curr = NULL;
		}		
	}
	pthread_mutex_unlock(&mutex);
	pthread_exit(NULL);
	pthread_mutex_lock(&mutex);
}
